package maxminSalary;


import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class MaxMinSalaryReducer extends Reducer<Text, Text, Text, NullWritable> {
    //用来存放最小最小值公司名
    private final Text maxSalaryCity = new Text("");
    //用来存放最大薪资值公司名
    private final Text minSalaryCity = new Text("");
    //存放最大薪资的Double类型的值
    private final DoubleWritable maxSalary = new DoubleWritable(0.0);
    //存放最小薪资的Double类型的值
    private final DoubleWritable minSalary = new DoubleWritable(0.0);
    //存放最大薪资的字符串类型类型的值
    private final Text maxSalaryStr = new Text("");
    //存放最小薪资的字符串类型类型的值
    private final Text minSalaryStr = new Text("");

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {;
        System.out.println(key.toString());
        for (Text value : values) {

            //该方法对每个公司的薪资字符串进行获取最右边的double值
            double leftMaxAndMinSalary = getLeftMax(value.toString());
            //寻找最大薪资和最小薪资
            if(leftMaxAndMinSalary > maxSalary.get()) {
                maxSalaryStr.set(value);
                maxSalaryCity.set(key);
                maxSalary.set(leftMaxAndMinSalary);
            }else {
                minSalaryStr.set(value);
                minSalaryCity.set(key);
                minSalary.set(leftMaxAndMinSalary);
            }
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
      //将最大薪资和对应公司名，最大薪资和对应公司名，拼接成字符串，输出
        String maxSalaryCityString = maxSalaryCity + "," + maxSalaryStr +"\n";
      String minSalaryCityString = minSalaryCity+ "," + minSalaryStr;
      context.write(new Text(maxSalaryCityString+minSalaryCityString), NullWritable.get());

    }

    //将薪资字符串转为最右边的double值
    //2-2.5万 调用这个方法后获得 （double）2.5
    private double getLeftMax(String salaryStr){
        String[] dataArray = salaryStr.split("-");
        String leftMaxSalary = dataArray[1].replace("万","");
        return Double.parseDouble(leftMaxSalary);
    }
}